// Copyright 2019 StreamSets Inc.
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package spooler

import (
	log "github.com/sirupsen/logrus"
	"github.com/spf13/cast"
	"github.com/streamsets/datacollector-edge/api"
	"github.com/streamsets/datacollector-edge/container/recordio/wholefilerecord"
)

func (s *SpoolDirSource) produceWholeFileRecord(batchMaker api.BatchMaker) {
	fInfo := s.spooler.getCurrentFileInfo()

	fileMetadata, err := wholefilerecord.GetFileInfo(fInfo.getFullPath())
	if err != nil {
		log.WithError(err).Error("Failed to get File metadata")
		s.GetStageContext().ReportError(err)
		return
	}

	fileRef := wholefilerecord.NewLocalFileRef(
		fInfo.getFullPath(),
		cast.ToInt64(s.Conf.DataFormatConfig.WholeFileMaxObjectLen),
		cast.ToInt(s.Conf.DataFormatConfig.RateLimit),
	)

	recordReaderFactory := s.Conf.DataFormatConfig.RecordReaderFactory
	recordReader, err := recordReaderFactory.CreateWholeFileReader(
		s.GetStageContext(),
		"whole_file",
		fileMetadata,
		fileRef,
	)
	if err != nil {
		log.WithError(err).Error("Failed to create record reader")
		s.GetStageContext().ReportError(err)
		return
	}

	if wholeFileRecord, err := recordReader.ReadRecord(); err != nil {
		s.GetStageContext().ReportError(err)
	} else {
		batchMaker.AddRecord(wholeFileRecord)
		s.spooler.getCurrentFileInfo().setOffsetToRead(EOFOffset)
	}
}
